òɾۿûѧϰʹá
ԭַhttps://www.joinquant.com/post/11736

ԭһ˵ʽ鵽ԭĺ߽ۡ


ԭĲԴ£

# ¡Ծۿ£https://www.joinquant.com/post/1810
# ⣺áЭİש
# ߣJoinQuant

import numpy as np
import pandas as pd

#===========================================

def initialize(context):
    set_params()
    set_variables()
    set_backtest()

# ---1. ò
def set_params():
    # Ʊ1
    g.security1 = '603288.XSHG' 
    # Ʊ2
    g.security2 = '600036.XSHG'
    # ׼
    g.benchmark = '600036.XSHG'
    # عϵ
    g.regression_ratio = 1.000#0.9574#0.9938
    # Ʊ1Ĭϲλ
    g.p = 0.5
    # Ʊ2Ĭϲλ
    g.q = 0.5
    # z-score
    g.test_days = 120

# ---2. ñ
def set_variables():
    # ״̬
    g.state = 'empty'

# ---3. ûز
def set_backtest():
    # û׼
    set_benchmark(g.benchmark)
    # ֻ
    log.set_level('order', 'error')
    # ʵ۸
    set_option('use_real_price', True) 
    # ޻
    set_slippage(FixedSlippage(0.))

#==============================================
    
# ÿλʱ(ز,ÿһ,,ÿӵһ)һ
def handle_data(context, data):
    new_state = get_signal()
    change_positions(new_state,context)

# ---4.z-score
def z_test():
    # ȡ֧Ʊʷ۸
    prices1 = np.array(attribute_history(g.security1, g.test_days, '1d', ['close']).close)
    prices2 = np.array(attribute_history(g.security2, g.test_days, '1d', ['close']).close)
    # ݻعǵƽ Y-a.X
    stable_series = prices2 - g.regression_ratio*prices1
    # ֵ
    series_mean = mean(stable_series)
    # ׼
    sigma = np.std(stable_series)
    # ֵֵ
    diff = stable_series[-1] - series_mean
    # zֵ
    return(diff/sigma)

# ---5.ȡź
# µ״̬һstring
def get_signal():
    z_score = z_test()
    if z_score > 1:
        # ״̬Ϊȫֵһ֧
        return('buy1')
    # Сڸ׼
    if z_score < -1:
        # ״̬Ϊȫֵڶ֧
        return('buy2')
    # ׼֮
    if -1 <= z_score <= 1:
        # 0
        if z_score >= 0:
            # ھֵ
            return('side1')
        # ֮
        else:
            # ھֵ
            return('side2')

# ---6.źŵλ
# Ŀ״̬Ϊһstring
def change_positions(new_state,context):
    # ֵֵ
    total_value = context.portfolio.portfolio_value
    # ״̬ȫֹƱ1
    if new_state == 'buy1':
        # ȫƱ2
        order_target(g.security2, 0)
        # ȫƱ1
        order_value(g.security1, total_value)
        # ״̬
        g.state = 'buy1'
    # ״̬ȫֹƱ2
    if new_state == 'buy2':
        # ȫƱ1
        order_target(g.security1, 0)
        # ȫƱ2
        order_value(g.security2, total_value)
        # ״̬
        g.state = 'buy2'
    # ȫһƱ״̬z-score0
    if (g.state == 'buy1' and new_state == 'side2') or (g.state == 'buy2' and new_state == 'side1'):
        # p,qֵƱλΪĬֵ
        order_target_value(g.security1, g.p * total_value)
        order_target_value(g.security2, g.q * total_value)
        # ظΪҪ򣬶ûصȷĸĸ
        order_target_value(g.security1, g.p * total_value)
        order_target_value(g.security2, g.q * total_value)
        # ״̬Ϊƽ
        g.state = 'even'
